package com.irt.rootcloud.realtime.gketl;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;

import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHits;

public class test {

	public static void main(String args[]) {
//		writeEs("d:\\2.json","gps","dis_map");
		writeEs(args[0],args[1],args[2]);
	}

	public static FileInputStream getFileInputStream(String path) {
		FileInputStream fis = null;
		try {
			File pfile = new File(path);
			fis = new FileInputStream(pfile);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return fis;
	}

	public static void etljson() {

	}

	public static void readEs() {
		Settings settings = ImmutableSettings.settingsBuilder()
				.put("cluster.name", "saas-dev-es").build();
		Client client = new TransportClient(settings)
				.addTransportAddress(new InetSocketTransportAddress(
						"10.70.17.205", 9300));
		SearchResponse response = client.prepareSearch("alias1_test")
				.setTypes("1001").setQuery(QueryBuilders.matchAllQuery())
				.setSize(10000).setScroll(new TimeValue(600000))
				.setSearchType(SearchType.SCAN).execute().actionGet();// setSearchType(SearchType.Scan)
																		// 告诉ES不需要排序只要结果返回即可
																		// setScroll(new
																		// TimeValue(600000))
																		// 设置滚动的时间
		String scrollid = response.getScrollId();
		try {
			// 把导出的结果以JSON的格式写到文件里
			BufferedWriter out = new BufferedWriter(new FileWriter("es", true));

			// 每次返回数据10000条。一直循环查询直到所有的数据都查询出来
			while (true) {
				SearchResponse response2 = client.prepareSearchScroll(scrollid)
						.setScroll(new TimeValue(1000000)).execute()
						.actionGet();
				SearchHits searchHit = response2.getHits();
				// 再次查询不到数据时跳出循环
				if (searchHit.getHits().length == 0) {
					break;
				}
				System.out.println("查询数量 ：" + searchHit.getHits().length);
				for (int i = 0; i < searchHit.getHits().length; i++) {
					String json = searchHit.getHits()[i].getSourceAsString();
					out.write(json);
					out.write("\r\n");
				}
			}
			System.out.println("查询结束");
			out.close();
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public static void writeEs(String path,String ind,String type) {
		Settings settings = ImmutableSettings.settingsBuilder()
				.put("cluster.name", "saas-dev-es").build();
		Client client = new TransportClient(settings)
				.addTransportAddress(new InetSocketTransportAddress(
						"10.70.17.205", 9300));
		try {
			// 读取刚才导出的ES数据
			BufferedReader br = new BufferedReader(new FileReader(path));
			String json = null;
			int count = 0;
			// 开启批量插入
			BulkRequestBuilder bulkRequest = client.prepareBulk();
			while ((json = br.readLine()) != null) {
				count++;
				bulkRequest.add(client.prepareIndex(ind, type).setSource(json));
//				// 每一千条提交一次
				if (count % 10000 == 0) {
					bulkRequest.execute().actionGet();
					System.out.println("提交了：" + count);
					bulkRequest = client.prepareBulk();
				}
			}
			bulkRequest.execute().actionGet();
			System.out.println("插入完毕");
			br.close();
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
